package com.atguigu.flink.wordcount;

import com.atguigu.flink.pojo.WordCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 *
 *      使用lamda表达式简化
 *
 *      Caused by: org.apache.flink.api.common.functions.InvalidTypesException:  非法类型异常
 *                  The generic type parameters of 'Collector' are missing.  Collector的泛型参数缺失。 泛型只在编译时有效！编译后统一替换为Object,这个行为也称为泛型擦除。
 *
 *                  很多情况下，lamda式中如果有Java的泛型，不会提供足够的提取泛型类型的信息
 *                  In many cases lambda methods don't provide enough information for automatic type extraction
 *                  when Java generics are involved.
 *
 *                  如果避免此类错误，可以使用匿名内部类
 *                  An easy workaround is to use an (anonymous) class instead that implements the 'org.apache.flink.api.common.functions.FlatMapFunction' interface.
 *
 *                  如果坚持使用lamda表达式，需要明确显式声明泛型的信息
 *                 Otherwise the type has to be specified explicitly using type information.
 *
 *                  可以在转换的结果的后面使用 returns(...) 提供类型的信息。
 *                 You can give type information hints by using the returns(...) method on the result of the transformation call
 *
 *      -----------------
 *               Collector<Tuple2<String, Integer>>      Collector<WordCount> out
 *          类型：  Collector                            Collector
 *          泛型:   Tuple2<String, Integer>              WordCount
 */
public class Demo6_LamdaPOJO
{
    public static void main(String[] args) throws Exception {

        System.out.println("hello main....");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        env.setParallelism(1);
        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888);

        ds.flatMap((FlatMapFunction<String, WordCount>) (value, out) -> {
            String[] words = value.split(" ");
            for (String word : words) {
                out.collect(new WordCount(word,1));
            }
            // 明确声明Collector其中的泛型是WordCount类型
        }).returns(WordCount.class)
          //可以使用 方法引用 简化
          // 方法引用： 在lamda表达式中，实现的接口的抽象方法中的逻辑，正好和某个类中已经实现的方法一模一样。直接引用这个类的这个方法
          .keyBy(WordCount::getWord)
          .sum("count")
          .print();


        env.execute();

    }


}
